package com.atguigu.flink.streamapi.exec;

import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/11/22
 *
 *      搞清楚输入和输出。
 *          由输入正推理 输出
 *          由输出倒推理 输入
 *
 *      543462,1715,1464116,pv,1511658000
 * userid,访问的页面项的id,访问内容的分类id,行为类型,ts
 *
 *      每得到一行pv,记1，累加
 */
public class Demo1_PV
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.readTextFile("data/UserBehavior.csv")
           .flatMap(new FlatMapFunction<String, UserBehavior>()
           {
               @Override
               public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                   String[] words = value.split(",");
                   if ("pv".equals(words[3])){
                       out.collect(new UserBehavior(
                            Long.valueOf(words[0]),
                            Long.valueOf(words[1]),
                            Integer.valueOf(words[2]),
                           words[3],
                            Long.valueOf(words[4])
                       ));
                   }
               }
           })
           .process(new ProcessFunction<UserBehavior, String>()
           {
               int sum = 0;
               int sum1 = 0;

               @Override
               public void processElement(UserBehavior value, Context ctx, Collector<String> out) throws Exception {
                   sum += 1;
                   out.collect("pv:" + sum);
                   System.out.println(sum1);
               }
           }).setParallelism(1)
           .print().setParallelism(1);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
